package com.sheep.transer.processor;

import org.apache.activemq.command.ActiveMQTextMessage;
import org.apache.camel.Exchange;
import org.apache.camel.Handler;
import org.apache.camel.Processor;
import org.apache.camel.component.kafka.KafkaConstants;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Created with IntelliJ IDEA.
 * User: Administrator
 * Date: 16-6-13
 * Time: 上午11:15
 * To change this template use File | Settings | File Templates.
 */
public class ActiveMQ2KafkaProcessor implements Processor {
    private static Logger logger = LoggerFactory.getLogger(ActiveMQ2KafkaProcessor.class);
    private String id = "";
    private static long count = 0l;

    public String getId() {
        return id;
    }

    public void setId(String id) {
        this.id = id;
    }

    @Handler
    public void process(Exchange exchange) {
        try {
            ActiveMQTextMessage message = (ActiveMQTextMessage) exchange.getIn().getBody();
            if (message.getContent().getLength() > 4) {
                byte[] content = new byte[message.getContent().getLength() - 4];
                System.arraycopy(message.getContent().data, 4, content, 0, content.length);
                String me = new String(content);
                exchange.getIn().setBody(id+":"+me, String.class);
                exchange.getIn().setHeader(KafkaConstants.PARTITION_KEY, 0);
                exchange.getIn().setHeader(KafkaConstants.KEY, "1");
            }
        } catch (Exception e) {
            logger.error("信息转发失败" + e.getMessage());
        }
    }
}
